package service;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import pojo.LogBean;
import pojo.RuleAutomicParam;
import pojo.RuleParam;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * 从hbase里面查询用户属性的数据，看是否能够匹配上。
 */
public class UserProfileQueryServiceImpl implements UserProfileQueryService {
    Connection conn;
    Table htable;


    public UserProfileQueryServiceImpl() {
        org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
        conf.set("hbase.zookeeper.quorum", "master3:2181,master4:2181,master5:2181");
        try {
            conn = ConnectionFactory.createConnection(conf);
            htable = conn.getTable(TableName.valueOf("yiniu2"));
        } catch (IOException e) {
            throw new RuntimeException("hbase集群异常");
        }
}


    /**
     * TODO  我们查到的标签属性最好要吧结果放到缓存，提升效率。有待改造
     * @param logBean
     * @param ruleParam
     * @return
     * @throws IOException
     */

    public boolean QueryUserProfileQueryFromHbase(LogBean logBean, RuleParam ruleParam) throws IOException {
        //获取传入的规则参数
        RuleAutomicParam triggerparam = ruleParam.getTriggerparam();

        //获取当前的事件id
        String eventId = logBean.getEventId();
        if (eventId.equals(triggerparam.getEventId())) {
            //触发事件凡符合规则,就去hbase里面查询这个用户的画像
            HashMap<String, String> properties = ruleParam.getUserProfileParam();
            Set<Map.Entry<String, String>> entries = properties.entrySet();
            Set<String> keys = properties.keySet();

            Get get = new Get(logBean.getDeviceId().getBytes());
            //可能会有多个画像属性的判断,所以需要循环把列加在get的quqlifify上
            for (String key : keys) {
                get.addColumn("f".getBytes(), key.getBytes());
            }

            Result result = htable.get(get);
            //将结果进行比对
            for (Map.Entry<String, String> entry : entries) {
                String key = entry.getKey();
                String value = entry.getValue();
                //从查询结果中取出标签的值
                byte[] tagvalue = result.getValue("f".getBytes(), key.getBytes());
                //判断拿到的值和条件值,注意数组为空，这里的new String就有问题
                System.out.println(new String(tagvalue));
                if ( !new String(tagvalue).equals(value)) {//不一致，直接返false，一致就往下继续循环
                    return false;
                }
                //todo  这里不能这么写
//                else {
//                    return true;
//                }
            }

            //代码如果走到这里，就代表之前的所有的属性值都满足条件的值
            return true;
        }
        return false;
    }


}
